package com.apps.sdses.flink141.source;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;

import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;

/**
 * @author jiwei
 * @description geo数据生成器
 * @date 2023/6/14 23:13
 */
public class GeomDataSource extends RichSourceFunction<String> {
    private boolean isRunning = true;
    private final Random rand = new Random();
    List<String> carNos = new ArrayList<>(5);

    private final int messageNum = 100000;
    List<String> message = new ArrayList<>(messageNum);

    private final String carNosPrefix = "鲁A";
    private final double lonMin = 113.013984; //济南地理位置经度最小值
    private final double lonMax = 114.078879; //济南地理位置经度最大值
    private final double latMin = 22.540088;  //济南地理位置维度最小值
    private final double latMax = 22.707716;  //济南地理位置维度最大值

    private final SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd' 'HH:mm:ss.SSS");

    private final String dataFormat = "{\"geom_data\": \"POINT(%f %f)\", \"car_nos\": %s, \"msg_id\": %d, \"event_time\": \"%s\"}";


    @Override
    public void open(Configuration parameters) throws Exception {
        //初始化20条车牌号列表
//        for (int i = 0; i < 20; i++) {
//            int num = rand.nextInt(100000);
//            carNos.add(carNosPrefix + String.format("%05d", num));
//        }

//        //初始化5个车辆
        carNos.add(carNosPrefix + "52638");
        carNos.add(carNosPrefix + "12224");
        carNos.add(carNosPrefix + "33082");
        carNos.add(carNosPrefix + "30440");
        carNos.add(carNosPrefix + "60642");

    }

    @Override
    public void run(SourceContext<String> ctx) throws Exception {

//        for (int j = 0; j < 1000000; j++) {

//        }
        while (true) {
            //坐标
            double lon = Double.parseDouble(String.format("%.6f", lonMin + rand.nextDouble() * (lonMax - lonMin)));
            double lat = Double.parseDouble(String.format("%.6f", latMin + rand.nextDouble() * (latMax - latMin)));

            //随机5个的车牌号
            ArrayList<String> selectCarNos = new ArrayList<>();
            int carNoCount = rand.nextInt(5) + 1;
            for (int i = 0; i < carNoCount; i++) {
                String carNo = carNos.get(rand.nextInt(carNos.size()));
                selectCarNos.add(carNo);
            }
            //随机消息id
            long msgId = System.currentTimeMillis();

            String eventTime = sdf.format(System.currentTimeMillis());

            String geomJson = String.format(dataFormat,
                    lon, lat, String.format("[\"%s\"]", String.join("\", \"", selectCarNos)), msgId, eventTime
            );


            for (int i = 1; i <= messageNum; i++) {
                message.add(geomJson);
                if (i == messageNum) {
                    message.forEach( x -> ctx.collect(x));
//                    System.out.println("===================");
//                    System.out.println(message.size() + "------" + sdf.format(System.currentTimeMillis()));
//                    System.out.println("===================");
                    message.clear();
                    Thread.sleep(1000);
                }
            }




            // 输出数据
//            ctx.collect(geomJson);


//            Thread.sleep(1000);
        }
    }

    @Override
    public void cancel() {
        isRunning = false;
    }
}
